package com.cxl.mapreduce._8outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @author joyson
 * @create 2019-06-16-上午9:36
 */
public class FRecordWriter extends RecordWriter<Text, NullWritable> {
    FSDataOutputStream fscxl;
    FSDataOutputStream fsother;
    public FRecordWriter(TaskAttemptContext taskAttemptContext) {
        try {
            FileSystem fs = FileSystem.get(taskAttemptContext.getConfiguration());
            fscxl = fs.create(new Path("/Users/a123/devWorkspace/bigData/hdfs/src/main/resources/output/outputformat/cxl.log"));
            fsother = fs.create(new Path("/Users/a123/devWorkspace/bigData/hdfs/src/main/resources/output/outputformat/other.log"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        if(key.toString().contains("cxl")){
            fscxl.write(key.toString().getBytes());
        }else {
            fsother.write(key.toString().getBytes());
        }
    }

    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(fscxl);
        IOUtils.closeStream(fsother);
    }
}
